
package com.ksyun.campus.client;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ksyun.campus.client.common.constant.ServerConfiguration;
import com.ksyun.campus.client.domain.BaseResult;
import com.ksyun.campus.client.domain.ReplicaData;
import com.ksyun.campus.client.domain.StatInfo;
import com.ksyun.campus.client.domain.dto.req.ReadReqDto;
import com.ksyun.campus.client.domain.dto.resp.ReadRespDto;
import com.ksyun.campus.client.util.HttpClientConfig;
import com.ksyun.campus.client.util.HttpClientUtil;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.hc.client5.http.classic.HttpClient;
import org.apache.hc.client5.http.classic.methods.HttpPost;
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.http.io.entity.EntityUtils;
import org.apache.hc.core5.http.io.entity.StringEntity;

import java.io.IOException;
import java.io.InputStream;

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
@Slf4j
public class FSInputStream extends InputStream {
    private static final int CHUNK_SIZE = 2 * 1024 * 1024; // 20MB

    private StatInfo statInfo;
    private String path;
    private String defaultFileSystemName;
    private Integer offset = 0;

    @Override
    public int read() throws IOException {
        byte[] singleByte = new byte[1];
        int bytesRead = read(singleByte, 0, 1);
        return bytesRead == -1 ? bytesRead : (singleByte[0]) & 0xFF;
    }

    @Override
    public int read(byte[] b) throws IOException {
        return read(b, 0, b.length);
    }

    @Override
    public int read(byte[] b, int off, int len) throws IOException {
        if (b == null) {
            throw new NullPointerException();
        } else if ((off < 0) || (len < 0) || ((off + len) > b.length)) {
            throw new IndexOutOfBoundsException();
        } else if (len == 0) {
            return 0;
        }

        int totalBytesRead = 0;
        int remainingBytes = len;

        while (remainingBytes > 0) {
            int chunkSize = Math.min(CHUNK_SIZE, remainingBytes);
            int bytesRead = readChunk(b, off, chunkSize);

            if (bytesRead == -1) {
                break;
            }

            totalBytesRead += bytesRead;
            remainingBytes -= bytesRead;
            off += bytesRead;
        }

        return totalBytesRead == 0 && remainingBytes > 0 ? -1 : totalBytesRead;
    }

    private int readChunk(byte[] b, int off, int len) throws IOException {
        if (len > statInfo.getSize() - offset) {
            len = (int) (statInfo.getSize() - offset);
        }

        ReplicaData replicaData = statInfo.getReplicaData().get(0);
        String ipAndPort = replicaData.getDsNode();
        String dataServerUrl = ServerConfiguration.HTTP + ipAndPort + ServerConfiguration.DATAREADPATH;

        ReadReqDto readReqDto = ReadReqDto.builder()
                .path(path)
                .statInfo(statInfo)
                .offset(offset)
                .length(len)
                .build();

        HttpPost httpPost = new HttpPost(dataServerUrl);
        httpPost.addHeader("fileSystemName", defaultFileSystemName);
        ObjectMapper objectMapper = new ObjectMapper();
        String requestJson = objectMapper.writeValueAsString(readReqDto);
        StringEntity requestEntity = new StringEntity(requestJson, ContentType.APPLICATION_JSON);
        httpPost.setEntity(requestEntity);

        HttpClient httpClient = HttpClientUtil.createHttpClient(new HttpClientConfig());
        try (CloseableHttpResponse httpResponse = (CloseableHttpResponse) httpClient.execute(httpPost)) {
            int statusCode = httpResponse.getCode();
            if (statusCode == HttpStatus.SC_OK) {
                String responseJson = EntityUtils.toString(httpResponse.getEntity());
                BaseResult<ReadRespDto> response = objectMapper.readValue(responseJson, new TypeReference<BaseResult<ReadRespDto>>() {});
                ReadRespDto readRespDto = response.getData();

                if (readRespDto.getLength() == -1) {
                    return -1;
                }

                byte[] fileBytes = readRespDto.getData();
                offset += readRespDto.getLength();
                System.arraycopy(fileBytes, 0, b, off, readRespDto.getLength());

                log.info("Successfully read data from DataServer.");
                return readRespDto.getLength();
            } else {
                throw new IOException("Failed to read from DataServer with HTTP status code: " + statusCode);
            }
        } catch (Exception e) {
            log.error("Failed to read from DataServer", e);
            throw new IOException("Failed to read from DataServer", e);
        }
    }

    @Override
    public void close() throws IOException {
        super.close();
    }
}


//
//import com.fasterxml.jackson.core.type.TypeReference;
//import com.fasterxml.jackson.databind.ObjectMapper;
//import com.ksyun.campus.client.common.constant.ServerConfiguration;
//import com.ksyun.campus.client.domain.BaseResult;
//import com.ksyun.campus.client.domain.ReplicaData;
//import com.ksyun.campus.client.domain.StatInfo;
//import com.ksyun.campus.client.domain.dto.req.ReadReqDto;
//import com.ksyun.campus.client.domain.dto.resp.ReadRespDto;
//import com.ksyun.campus.client.util.HttpClientConfig;
//import com.ksyun.campus.client.util.HttpClientUtil;
//import lombok.AllArgsConstructor;
//import lombok.Builder;
//import lombok.Data;
//import lombok.NoArgsConstructor;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.hc.client5.http.classic.HttpClient;
//import org.apache.hc.client5.http.classic.methods.HttpPost;
//import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
//import org.apache.hc.core5.http.ContentType;
//import org.apache.hc.core5.http.HttpStatus;
//import org.apache.hc.core5.http.io.entity.EntityUtils;
//import org.apache.hc.core5.http.io.entity.StringEntity;
//
//import java.io.IOException;
//import java.io.InputStream;
//
//@Data
//@AllArgsConstructor
//@NoArgsConstructor
//@Builder
//@Slf4j
//public class FSInputStream extends InputStream {
//    private StatInfo statInfo;
//    private String path;
//    private String defaultFileSystemName;
//    /**
//     * 真实文件的当前偏移量
//     */
//    private Integer offset = 0;
//
//    @Override
//    public int read() throws IOException {
//        byte[] singleByte = new byte[1];
//        int bytesRead = read(singleByte, 0, 1);
//        //读到文件末尾则返回-1，否则返回输入流的下一个字节
//        return bytesRead == -1 ? bytesRead : (singleByte[0]) & 0xFF;
//    }
//
//    @Override
//    public int read(byte[] b) throws IOException {
//        return read(b, 0, b.length);
//    }
//
//
//    //成型的read方法
//    @Override
//    public int read(byte[] b, int off, int len) throws IOException {
//        //验证读是否合法
//        if (b == null) {
//            throw new NullPointerException();
//        } else if ((off < 0) || (len < 0) ||
//                ((off + len) > b.length)) {
//            throw new IndexOutOfBoundsException();
//        } else if (len == 0) {
//            return 0;
//        }
//        if(len>statInfo.getSize()){
//            len= (int) statInfo.getSize();
//        }
//        // 与dataServer通信获得文件内容
//        ReplicaData replicaData = statInfo.getReplicaData().get(0);
//        String ipAndPort = replicaData.getDsNode();
//        String dataServerUrl = ServerConfiguration.HTTP + ipAndPort + ServerConfiguration.DATAREADPATH;
//        ReadReqDto readReqDto = ReadReqDto.builder()
//                .path(path)
//                .statInfo(statInfo)
//                .offset(offset)
//                .length(len)
//                .build();
//        HttpPost httpPost = new HttpPost(dataServerUrl);
//        httpPost.addHeader("fileSystemName", defaultFileSystemName);
//        ObjectMapper objectMapper = new ObjectMapper();
//        String requestJson = objectMapper.writeValueAsString(readReqDto);
//        StringEntity requestEntity = new StringEntity(requestJson, ContentType.APPLICATION_JSON);
//        httpPost.setEntity(requestEntity);
////
////        HttpGet dataGet = new HttpGet(dataServerUrl);
////        dataGet.addHeader("fileSystemName", defaultFileSystemName);
//
//        HttpClient httpClient = HttpClientUtil.createHttpClient(new HttpClientConfig());
//        try (CloseableHttpResponse httpResponse = (CloseableHttpResponse) httpClient.execute(httpPost)) {
//            int statusCode = httpResponse.getCode();
//            if (statusCode == HttpStatus.SC_OK) {
//                String responseJson = EntityUtils.toString(httpResponse.getEntity());
//                BaseResult<ReadRespDto> response = objectMapper.readValue(responseJson, new TypeReference<BaseResult<ReadRespDto>>() {
//                });
//                ReadRespDto readRespDto = response.getData();
//                Integer readLength = readRespDto.getLength();
//                // 数据已经读完了
//                if (readLength == -1) {
//                    return readLength;
//                }
//                byte[] fileBytes = readRespDto.getData();
//                // 更新真实文件的偏移量
//                offset += readLength;
//                // 将接收到的byte数组复制给read方法的参数b(byte[])
//                System.arraycopy(fileBytes, 0, b, off, readLength);
////                log.info("成功从DataServer读取数据");
//                return readLength;
//            } else {
//                throw new IOException("Failed to read from DataServer with HTTP status code: " + statusCode);
//            }
//        } catch (Exception e) {
//            e.printStackTrace();
////            throw new IOException("Failed to read from DataServer", e);
//            // TODO: 2024/8/16 读没有成功的解决方式
//            return -1;
//        }
////        return super.read(b, off, len);
//    }
//
//    @Override
//    public void close() throws IOException {
//        super.close();
//    }
//}
